package HandshakeJoin;

import java.util.Calendar;
import java.net.URL;
import com.google.common.base.Preconditions;
import org.apache.s4.base.Event;
import org.apache.s4.core.adapter.AdapterApp;
import org.apache.s4.core.Stream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class InputAdapterApp extends AdapterApp
{
    transient private static final Logger logger = LoggerFactory
            .getLogger(InputAdapterApp.class);

    private String sourceFilename = null;
    private int emitInterval = 50;
    private int sizeOfOneEmit = 1;
    private int numFields = 1;
    private boolean randomSource = true;

    @Override
    protected void onInit() {
        setKeyFinder(new DataKeyFinder());
        super.onInit();

        /* Configurations */
        setGlobalParameters("/external_input.txt");

        SourceConverter sourceConvertor = createPE(SourceConverter.class);
        sourceConvertor.setOutputStream(getRemoteStream())
                .setSourceFilename(sourceFilename)
                .setSizeOfOneEmit(sizeOfOneEmit).setEmitInterval(emitInterval)
                .setSingleton(true);

        if (randomSource) {
            Stream<Event> dataStream = createStream("Data",
                    new DataKeyFinder(), sourceConvertor);

            GenerateDataPE generate = createPE(GenerateDataPE.class);
            generate.setDownStream(dataStream).setNumberOfFields(numFields)
                    .setRandomSeed(Calendar.getInstance().getTimeInMillis())
                    .setIngestInterval(emitInterval).setSingleton(true);
        }
    }

    @Override
    protected void onStart() {
        logGlobalParameters();
    }

    @Override
    protected void onClose() {}

    private void setGlobalParameters(final String filename) {
        URL url = this.getClass().getResource(filename);
        Preconditions.checkState(url != null, "invalid input description file");

        String streamName = getRemoteStream().getName();
        InputConfigure config = (new InputConfigure()).parseInputDespFile(url,
                streamName);

        sourceFilename = config.getSourceFilename();
        emitInterval = config.getEmitInterval();
        sizeOfOneEmit = config.getSizeOfOneEmit();
        randomSource = config.getEnableRandomSource();
        numFields = config.getNumberOfFieldsPerTuple();
    }

    private void logGlobalParameters() {
        logger.info("Stream name: " + getRemoteStream().getName());
        logger.info("Source file name: " + sourceFilename);
        logger.info("Emit interval: " + emitInterval);
        logger.info("Size of every emitting: " + sizeOfOneEmit);
        logger.info("Enable random source: " + randomSource);
        logger.info("Number of fields in random tuple: " + numFields);
    }
}